/**
 * 
 */
package edu.umd.clip.lm.model.training;

import java.io.*;
import java.lang.ref.*;
import java.nio.*;
import java.util.*;

import edu.umd.clip.lm.util.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
class Word2ClustersCounts implements Serializable {
	private static final long serialVersionUID = 1L;

	private final File file;

	private transient Writing writing;
	private transient Reading reading;
	
	private static class Writing {
		DetachableOutputStream output;
	}
	
	private static class Reading {
		Reference<CompactReadOnlyInt2IntHashMap> cluster2count;
	}
	
	public Word2ClustersCounts(File file) {
		this.file = file;
	}
	
	public void openForWriting(Observer fileCache) {
		writing = new Writing();
		if (file.isFile()) {
			// truncate the file
			try {
				new FileOutputStream(file).close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		writing.output = new DetachableOutputStream(file, 8092, 4096);		
		writing.output.addFileOpenListener(fileCache);
	}
	
	public void addClusterCount(int cluster, int count) {
		ByteBuffer buf = ByteBuffer.allocate(8);
		buf.putInt(cluster);
		buf.putInt(count);
		
		try {
			writing.output.write(buf.array());
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public synchronized void finishWriting() {
		if (writing != null) {
			try {
				writing.output.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			writing = null;
		}
	}
	
	private CompactReadOnlyInt2IntHashMap read() throws IOException {
		RandomAccessFile raf = new RandomAccessFile(file, "r");
		ByteBuffer buf = ByteBuffer.allocate((int) raf.length());
		raf.getChannel().read(buf);
		raf.close();
		
		buf.flip();
		int numClusters = buf.limit() / 8;
		
		int clusters[] = new int[numClusters];
		int counts[] = new int[numClusters];
		
		int pos = 0;
		while(buf.hasRemaining()) {
			int cluster = buf.getInt();
			clusters[pos] = cluster;
			
			int count = buf.getInt();
			counts[pos] = count;
			
			++pos;
		}
		assert(pos == numClusters);
		buf = null;
		
		return new CompactReadOnlyInt2IntHashMap(clusters, counts);
	}
	
	public CompactReadOnlyInt2IntHashMap getClusterCounts() {
		if (reading == null) {
			synchronized(this) {
				if (reading == null) {
					Reading reading = new Reading();
					reading.cluster2count = new SoftReference<CompactReadOnlyInt2IntHashMap>(null);
					this.reading = reading;
				}
			}
		}
		CompactReadOnlyInt2IntHashMap counts = reading.cluster2count.get();
		if (counts == null) {
			synchronized(this) {
				counts = reading.cluster2count.get();
				if (counts == null) {
					try {
						counts = read();
						reading.cluster2count = new SoftReference<CompactReadOnlyInt2IntHashMap>(counts);
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			}
		}
		return counts;
	}
	
	public void clear() {
		file.delete();
	}
}
